package com.chen.q3.handler;

import com.chen.q3.filter.HttpResponseFilter;
import com.chen.q3.filter.impl.SimpleHttpResponseFilter;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.handler.codec.http.*;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;
import org.apache.lucene.util.NamedThreadFactory;

import java.io.IOException;
import java.util.concurrent.*;

public class HttpOutboundHandler extends ChannelOutboundHandlerAdapter {

    private CloseableHttpAsyncClient httpClient;
    private String proxiedUrl;
    private ExecutorService proxyService;
    private HttpResponseFilter responseFilter = new SimpleHttpResponseFilter();

    HttpOutboundHandler(String proxiedUrl){
        this.proxiedUrl = this.formatUrl(proxiedUrl);

        int cores = Runtime.getRuntime().availableProcessors();
        long keepAliveTime = 1000;
        int queueSize = 2048;
        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
        proxyService = new ThreadPoolExecutor(cores,cores,
                keepAliveTime, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(queueSize),
                new NamedThreadFactory("proxyService"),handler);
        RequestConfig requestConfig = RequestConfig.custom()
                .setConnectTimeout(20000)
                .setConnectionRequestTimeout(10000)
                .setSocketTimeout(10000)
                .build();
        //配置IO线程
        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(10)
                .setSoKeepAlive(true)
                .build();
        //设置连接池大小
        ConnectingIOReactor ioReactor = null;

        try {
            ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
        } catch (IOReactorException e) {
            e.printStackTrace();
        }
        PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
        connManager.setMaxTotal(40);
        connManager.setDefaultMaxPerRoute(8);

        httpClient = HttpAsyncClients.custom()
                .setConnectionManager(connManager)
                .setDefaultRequestConfig(requestConfig)
                .build();

        httpClient.start();
    }

    private String formatUrl(String proxiedUrl) {
        return proxiedUrl.endsWith("/")?proxiedUrl.substring(0,proxiedUrl.length()-1):proxiedUrl;
    }


    public void handle(final FullHttpRequest fullRequest, final ChannelHandlerContext ctx) {
        String url = proxiedUrl + fullRequest.uri();
        proxyService.submit(() -> fetchGet(fullRequest,ctx,url));

        //fetchGet(fullRequest,ctx,url);
    }

    private void fetchGet(FullHttpRequest inboundReq, ChannelHandlerContext ctx, String url) {
        System.out.println("请求转发URL："+url);
        final HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
        httpClient.execute(httpGet, new FutureCallback<HttpResponse>() {
            @Override
            public void completed(HttpResponse endpointResponse) {
                try {
                    System.out.println("请求"+url+"成功");
                    handleResponse(inboundReq,ctx,endpointResponse);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void failed(Exception e) {
                System.out.println("请求失败");
                httpGet.abort();
                ctx.close();
                e.printStackTrace();
            }

            @Override
            public void cancelled() {
                System.out.println("请求取消");
                httpGet.abort();
            }
        });
    }

    private void handleResponse(FullHttpRequest inboundReq, ChannelHandlerContext ctx, HttpResponse endpointResponse) {
        FullHttpResponse response = null;
        try {
            byte[] body = EntityUtils.toByteArray(endpointResponse.getEntity());

            response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(body));
            response.headers().set("Content-Type","text/html");
            response.headers().set("Content-Length",Integer.parseInt(endpointResponse.getFirstHeader("Content-Length").getValue()));

            responseFilter.filter(response);
            System.out.println(response.headers().get("response_filter"));
            long inboundTimestamp = Long.valueOf(inboundReq.headers().get("inbound_timestamp"));
            long outboundTimestamp = Long.valueOf(response.headers().get("outbound_timestamp")) ;
            long timeTotal = outboundTimestamp - inboundTimestamp;
            response.headers().set("req_seconds",(double)timeTotal/1000);
        } catch (IOException e) {
            e.printStackTrace();
            response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,HttpResponseStatus.NO_CONTENT);
            exceptionCaught(ctx,e);
        } finally {
            if (inboundReq != null ){
                if (!HttpUtil.isKeepAlive(inboundReq)){
                    ctx.write(response).addListener(ChannelFutureListener.CLOSE);
                }else {
                    ctx.write(response);
                }
            }
            ctx.flush();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}
